package kafkax

import (
	"context"
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/go-kratos/kratos/v2/log"
	"go-kratos-demo/internal/conf"
	"strings"
)

type KafkaConsumer struct {
	consumer *kafka.Consumer
	cmap     *kafka.ConfigMap
	log      *log.Helper
}

func NewKafkaConsumer(conf *conf.Data, logger log.Logger) (*KafkaConsumer, error) {
	log_ := log.NewHelper(log.With(logger, "x_module", "data/NewKafkaConsumer"))

	cMap := &kafka.ConfigMap{
		"bootstrap.servers":  strings.Join(conf.Kafka.Addresses, ","),
		"group.id":           conf.Kafka.DefaultGroup,
		"auto.offset.reset":  "earliest",
		"enable.auto.commit": true,
		"session.timeout.ms": 6000,
	}

	kc := &KafkaConsumer{
		cmap: cMap,
		log:  log_,
	}

	return kc, nil
}
func (c *KafkaConsumer) Subscribe(group string, topic string) (*KafkaConsumer, error) {
	err := c.cmap.SetKey("group.id", group)
	if err != nil {
		return nil, fmt.Errorf("[boot] failed SetKey group.id: %v", err)

	}
	consumer, err := kafka.NewConsumer(c.cmap)
	if err != nil {
		return nil, fmt.Errorf("[boot] failed opening connection to consumer: %v", err)
	}

	err = c.consumer.SubscribeTopics([]string{topic}, nil)
	if err != nil {
		return nil, fmt.Errorf("[boot] failed Subscribe topic  : %v", err)
	}
	c.consumer = consumer
	return c, nil
}

func (c *KafkaConsumer) Consume(ctx context.Context) {

	for {
		select {
		case ev := <-c.consumer.Events():
			switch e := ev.(type) {
			case *kafka.Message:
				c.log.Infof("Received message on topic %s partition %d offset %v: %s = %s",
					e.TopicPartition.Topic,
					e.TopicPartition.Partition,
					e.TopicPartition.Offset,
					string(e.Key),
					string(e.Value))
			case kafka.Error:
				if e.IsFatal() {
					c.log.Fatalf("Fatal error received from consumer: %v", e)
				} else {
					c.log.Errorf("Error received from consumer: %v", e)
				}
			case kafka.AssignedPartitions:
				c.log.Infof("Assigned partitions: %v", e)
			case kafka.RevokedPartitions:
				c.log.Infof("Revoked partitions: %v", e)
			}
		case <-ctx.Done():
			c.consumer.Close()
			return
		}
	}
}
